package com.mingte.manghe.utils;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author ：luxinsheng
 * @Email ：282860642@qq.com
 * @date ：Created in 2021/12/23 16:39
 * @description：并行处理工具类
 * @modified By：
 * @version: $
 */
public class TaskDisposeUtils {

    private static final int POOL_SIZE;

    /**
     * 当前jvm可用线程
     */
    static {
        POOL_SIZE = Math.max(Runtime.getRuntime().availableProcessors(),4);
    }

    /**
     * 并行处理
     * @param pool 线程池,可以为null
     * @param tasks 任务队列
     * @param consumer 消费者
     * @param <T>
     */
    public static <T> void disposeTask(ExecutorService pool, List<T> tasks, Consumer<T> consumer){
        disposeTask(true,POOL_SIZE,pool,tasks,consumer);
    }

    /**
     * 如果是自定义的线程池,进行销毁
     * @param moreThread
     * @param poolSize
     * @param pool
     * @param tasks
     * @param consumer
     * @param <T>
     */
    private static <T> void disposeTask(boolean moreThread,int poolSize,ExecutorService pool, List<T> tasks, Consumer<T> consumer){
        boolean threadPoolCustomize = false;
        if(tasks.isEmpty()){
            return;
        }
        if(null == pool){
            pool = Executors.newFixedThreadPool(poolSize);
            threadPoolCustomize = true;
        }
        if(moreThread && poolSize > 1){
            try {
                int countDownSize = Math.min(poolSize, tasks.size());
                CountDownLatch countDownLatch = new CountDownLatch(countDownSize);
                for (T item:tasks) {
                    pool.execute(()->{
                        try {
                            consumer.accept(item);
                        }finally {
                            countDownLatch.countDown();
                        }
                    });
                }
                countDownLatch.await();
            } catch (InterruptedException interruptedException) {
                interruptedException.printStackTrace();
            } finally {
                if(threadPoolCustomize && null != pool){
                    pool.shutdown();
                }
            }
        }else{
            for (T item:tasks) {
                consumer.accept(item);
            }
        }
    }

    /*public static void main(String[] args) {
        long startMain = System.currentTimeMillis();
        List<Integer> collect = Stream.iterate(1, a -> a + 1).limit(10).collect(Collectors.toList());
        TaskDisposeUtils.disposeTask(null,collect,(item)->{
            long start = System.currentTimeMillis();
            try {
                TimeUnit.SECONDS.sleep(item);
            } catch (InterruptedException interruptedException) {
                interruptedException.printStackTrace();
            }
            long end = System.currentTimeMillis();
            System.out.println(Thread.currentThread().getName() + " : " + System.currentTimeMillis() + " : 耗时"+(end-start) );
        });
        long endMain = System.currentTimeMillis();
        System.out.println(Thread.currentThread().getName() + " : " + System.currentTimeMillis() + " : 最终总耗时"+(endMain-startMain) );
    }*/

}
